package com.jfireframework.jnet.common.processor.worker;

import java.util.concurrent.ExecutorService;
import com.jfireframework.baseutil.reflect.UNSAFE;
import com.jfireframework.jnet.common.api.ChannelContext;
import com.jfireframework.jnet.common.api.ProcessorInvoker;
import com.jfireframework.jnet.common.util.FixArray;
import com.jfireframework.jnet.common.util.SPSCFixArray;

public class ChannelAttachWorker implements Runnable
{
    
    private static final int                    IDLE           = -1;
    private static final int                    WORK           = 1;
    private static final int                    SPIN_THRESHOLD = 1 << 7;
    private static final long                   STATE_OFFSET   = UNSAFE.getFieldOffset("state", ChannelAttachWorker.class);
    private final ExecutorService               executorService;
    private final FixArray<ChannelAttachEntity> entities       = new SPSCFixArray<ChannelAttachEntity>(512) {
                                                                   
                                                                   @Override
                                                                   protected ChannelAttachEntity newInstance()
                                                                   {
                                                                       return new ChannelAttachEntity();
                                                                   }
                                                               };
    
    private int                                 state          = IDLE;
    
    public ChannelAttachWorker(ExecutorService executorService)
    {
        this.executorService = executorService;
    }
    
    @Override
    public void run()
    {
        int spin = 0;
        do
        {
            long avail = entities.nextAvail();
            if (avail == -1)
            {
                spin = 0;
                for (;;)
                {
                    
                    if ((avail = entities.nextAvail()) != -1)
                    {
                        break;
                    }
                    else if ((spin += 1) < SPIN_THRESHOLD)
                    {
                        ;
                    }
                    else
                    {
                        state = IDLE;
                        if (entities.isEmpty() == false)
                        {
                            tryExecute();
                        }
                        return;
                    }
                }
            }
            ChannelAttachEntity slot = entities.getSlot(avail);
            ChannelContext channelContext = slot.channelContext;
            ProcessorInvoker invoker = slot.invoker;
            Object data = slot.data;
            slot.clear();
            try
            {
                entities.comsumeAvail(avail);
                if (invoker.process(data) == false)
                {
                    throw new IllegalStateException();
                }
            }
            catch (Throwable e)
            {
                e.printStackTrace();
                channelContext.close(e);
            }
        } while (true);
    }
    
    public boolean commit(ChannelContext channelContext, ProcessorInvoker invoker, Object data)
    {
        long offerIndexAvail = entities.nextOfferIndex();
        if (offerIndexAvail == -1)
        {
            return false;
        }
        ChannelAttachEntity slot = entities.getSlot(offerIndexAvail);
        slot.channelContext = channelContext;
        slot.invoker = invoker;
        slot.data = data;
        entities.commit(offerIndexAvail);
        tryExecute();
        return true;
    }
    
    private void tryExecute()
    {
        int now = state;
        if (now == IDLE && UNSAFE.compareAndSwapInt(this, STATE_OFFSET, IDLE, WORK))
        {
            executorService.execute(this);
        }
    }
    
    class ChannelAttachEntity
    {
        ChannelContext   channelContext;
        ProcessorInvoker invoker;
        Object           data;
        
        void clear()
        {
            channelContext = null;
            invoker = null;
            data = null;
        }
    }
}
